package com.als;

import com.alibaba.fastjson2.JSONObject;
import com.als.config.DatabaseConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.spark.ml.evaluation.RegressionEvaluator;
import org.apache.spark.ml.recommendation.ALS;
import org.apache.spark.ml.recommendation.ALSModel;
import org.apache.spark.sql.*;
import org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions;
import org.apache.spark.sql.expressions.Window;
import org.apache.spark.sql.expressions.WindowSpec;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.Statement;
import java.util.*;

import static org.apache.spark.sql.functions.*;

/**
 * ./spark-submit --class  com.als.ALSRecommendationSystem  --master spark://192.40.10.130:7077   /root/myspark/jars/als-1.0-SNAPSHOT.jar
 */
@Slf4j
public class ALSRecommendationSystem {

    private final SparkSession spark;
    private ALSModel model;
    private final DatabaseConfig globalDbConfig = YamlConfigReader.loadDatabaseConfig("application.yml");

    public ALSRecommendationSystem() {
        // 初始化 Spark 会话
        spark = SparkSession.builder().appName("ALS Recommendation System")
//                .master("local[*]")
                .master("local[1]").config("spark.sql.warehouse.dir", "target/spark-warehouse").config("spark.ui.port", "4040") // 固定端口:http://192.40.10.130:14040/
                .getOrCreate();

        // 打印所有配置确认UI设置
        Object configs = spark.conf().getAll();
        log.info("spark所有配置确认UI设置:{}", JSONObject.toJSONString(configs));

        // 获取实际UI地址（可能自动跳转端口）
        log.info("Spark UI:{}", spark.sparkContext().uiWebUrl());
    }

    /**
     * 主方法 - 示例用法
     */
    public static void main(String[] args) {
        ALSRecommendationSystem alsSystem = new ALSRecommendationSystem();

        try {
            // 1. 创建示例数据
          /*  Dataset<Row> ratings = alsSystem.createSampleData();
           log.info("示例数据：");
            ratings.show();*/

            // 从数据库读取数据，而不是硬编码
            Dataset<Row> ratings = alsSystem.readDataFromDatabase();

            if (ratings.isEmpty()) {
                log.info("数据库中没有数据，使用示例数据");
                ratings = alsSystem.createSampleData();
            }

            log.info("训练数据：");
            ratings.show(10);

            // 2. 训练模型
            log.info("开始训练模型...");
            alsSystem.trainModel(ratings);

            // 3. 为用户生成推荐
            alsSystem.recommendForUser(1, 3);
            alsSystem.recommendForUserAndSave(1, 3);
            alsSystem.recommendForAllUsersAndSave(3);

            // 4. 为物品推荐用户
            alsSystem.recommendUsersForItem(3, 2);
            alsSystem.recommendUsersForItemAndSave(3, 2);
            alsSystem.recommendUsersForAllItemsAndSave(2);

            // 5. 预测评分
            alsSystem.predictRating(2, 2);

            // 6. 保存模型
            alsSystem.saveModel("models/als_model");


            // 保持UI运行（等待用户输入后关闭）
            log.info("保持UI运行（等待用户输入后关闭）Press Enter to exit...");
            new Scanner(System.in).nextLine();

        } catch (Exception e) {
            log.error("ALS:{}", e.getMessage());
        } finally {
            alsSystem.spark.stop();
        }
    }

    /**
     * 创建示例数据
     */
    public Dataset<Row> createSampleData() {
        // 定义数据结构：用户ID，物品ID，评分
        StructType schema = new StructType(new StructField[]{DataTypes.createStructField("userId", DataTypes.IntegerType, false), DataTypes.createStructField("itemId", DataTypes.IntegerType, false), DataTypes.createStructField("rating", DataTypes.FloatType, false)});

        // 创建示例评分数据
        List<Row> data = Arrays.asList(RowFactory.create(1, 1, 5.0f), RowFactory.create(1, 2, 3.0f), RowFactory.create(1, 3, 4.0f), RowFactory.create(2, 1, 4.0f), RowFactory.create(2, 3, 2.0f), RowFactory.create(2, 4, 3.0f), RowFactory.create(3, 2, 1.0f), RowFactory.create(3, 3, 5.0f), RowFactory.create(3, 4, 4.0f), RowFactory.create(4, 1, 3.0f), RowFactory.create(4, 2, 4.0f), RowFactory.create(4, 4, 5.0f));

        return spark.createDataFrame(data, schema);
    }

    /**
     * 训练 ALS 模型
     */
    public void trainModel(Dataset<Row> ratings) {
        // 将数据分为训练集和测试集
        Dataset<Row>[] splits = ratings.randomSplit(new double[]{0.8, 0.2});
        Dataset<Row> training = splits[0];
        Dataset<Row> test = splits[1];

        // 配置 ALS 算法参数
        ALS als = new ALS().setMaxIter(10)                    // 最大迭代次数
                .setRegParam(0.01)                 // 正则化参数
                .setUserCol("userId")              // 用户列名
                .setItemCol("itemId")              // 物品列名
                .setRatingCol("rating")            // 评分列名
                .setColdStartStrategy("drop")      // 冷启动策略
                .setNonnegative(true);             // 非负约束

        // 训练模型
        model = als.fit(training);

        // 在测试集上评估模型
        Dataset<Row> predictions = model.transform(test);

        // 计算 RMSE
        RegressionEvaluator evaluator = new RegressionEvaluator().setMetricName("rmse").setLabelCol("rating").setPredictionCol("prediction");

        double rmse = evaluator.evaluate(predictions);
        log.info("模型 RMSE: " + rmse);
    }

    /**
     * 为用户生成推荐
     */
    public void recommendForUser(int userId, int numRecommendations) {
        if (model == null) {
            log.info("请先训练模型！");
            return;
        }

        // 为指定用户生成推荐
        Dataset<Row> userRecs = model.recommendForAllUsers(numRecommendations);

        // 过滤出指定用户的推荐
        Dataset<Row> userSpecificRecs = userRecs.filter(userRecs.col("userId").equalTo(userId));

        log.info("用户 " + userId + " 的推荐结果：");
        userSpecificRecs.show(false);
    }

    /**
     * 为所有用户生成推荐并批量写入MySQL
     */
    public void recommendForAllUsersAndSave(int numRecommendations) {
        if (model == null) {
            log.info("请先训练模型！");
            return;
        }

        try {
            DatabaseConfig dbConfig = YamlConfigReader.loadDatabaseConfig("application.yml");

            // 为所有用户生成推荐
            Dataset<Row> allUserRecs = model.recommendForAllUsers(numRecommendations);

            log.info("开始处理 {} 个用户的推荐结果...", allUserRecs.count());

            // 转换格式并写入数据库
            Dataset<Row> formattedRecs = allUserRecs.select(
                            col("userId"),
                            explode(col("recommendations"))
                                    .as("rec")).
                    select(
                            col("userId").as("user_id"),
                            col("rec.itemId").as("item_id"),
                            col("rec.rating").as("predicted_rating"),
                            current_timestamp().as("create_time"))
                    .withColumn("ranking", row_number().
                            over(Window.partitionBy("user_id").
                                    orderBy(col("predicted_rating").desc())));

            // 写入MySQL（覆盖模式）
            formattedRecs.write().format("jdbc").
                    option("url", dbConfig.getUrl()).
                    option("dbtable", "user_recommendations").
                    option("user", dbConfig.getUsername()).
                    option("password", dbConfig.getPassword()).
                    option("driver", dbConfig.getDriver()).
                    mode(SaveMode.Overwrite)  // 覆盖模式，清空旧数据
                    .save();

            log.info("成功为所有用户生成推荐，共 {} 条记录", formattedRecs.count());

            //或者调用保存方法saveRecommendationsWithProperties

        } catch (Exception e) {
            log.error("批量生成推荐失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 使用Properties配置写入推荐结果
     */
    public void saveRecommendationsWithProperties(Dataset<Row> recommendations) {
        DatabaseConfig dbConfig = YamlConfigReader.loadDatabaseConfig("application.yml");
        Properties connectionProperties = new Properties();
        connectionProperties.setProperty("user", dbConfig.getUsername());
        connectionProperties.setProperty("password", dbConfig.getPassword());
        connectionProperties.setProperty("driver", dbConfig.getDriver());
        connectionProperties.setProperty("batchsize", "1000");  // 批量大小
        connectionProperties.setProperty("isolationLevel", "READ_COMMITTED");

        recommendations.write().jdbc(dbConfig.getUrl(), "user_recommendations", connectionProperties);
    }

    /**
     * 为用户生成推荐并写入MySQL数据库
     */
    public void recommendForUserAndSave(int userId, int numRecommendations) {
        if (model == null) {
            log.info("请先训练模型！");
            return;
        }

        try {
            // 为指定用户生成推荐
            Dataset<Row> userRecs = model.recommendForAllUsers(numRecommendations);

            // 过滤出指定用户的推荐
            Dataset<Row> userSpecificRecs = userRecs.filter(userRecs.col("userId").equalTo(userId));

            // 解析推荐结果并写入数据库
            saveRecommendationsToMySQL(userSpecificRecs, userId);

        } catch (Exception e) {
            log.error("为用户 {} 生成推荐失败: {}", userId, e.getMessage(), e);
        }
    }

    /**
     * 将推荐结果保存到MySQL数据库
     */
    private void saveRecommendationsToMySQL(Dataset<Row> recommendations, int userId) {
        try {
            // 读取数据库配置
            DatabaseConfig dbConfig = YamlConfigReader.loadDatabaseConfig("application.yml");

            // 创建推荐结果表（如果不存在）
            createRecommendationTableIfNotExists(dbConfig);

            // 转换推荐结果格式
            Dataset<Row> formattedRecs = formatRecommendations(recommendations, userId);

            // 写入MySQL
            formattedRecs.write().format("jdbc").option("url", dbConfig.getUrl()).option("dbtable", "user_recommendations").option("user", dbConfig.getUsername()).option("password", dbConfig.getPassword()).option("driver", dbConfig.getDriver()).mode(SaveMode.Append)  // 追加模式
                    .save();

            log.info("用户 {} 的 {} 条推荐结果已保存到MySQL", userId, formattedRecs.count());

        } catch (Exception e) {
            log.error("保存推荐结果到MySQL失败: {}", e.getMessage(), e);
            throw new RuntimeException("保存推荐结果失败", e);
        }
    }

    /**
     * 转换推荐结果格式
     */
    private Dataset<Row> formatRecommendations(Dataset<Row> recommendations, int userId) {
        return recommendations.select(
                        col("userId"),
                        explode(col("recommendations")).as("rec"))
                .select(
                        col("userId"),
                        col("rec.itemId").as("itemId"),
                        col("rec.rating").as("predictedRating"),
                        current_timestamp().as("createTime")).
                withColumn("rank", monotonically_increasing_id())  // 添加排名
                .filter(col("userId").equalTo(userId));  // 再次过滤确保只有目标用户
    }

    /**
     * 创建推荐结果表（如果不存在）
     */
    private void createRecommendationTableIfNotExists(DatabaseConfig dbConfig) {
        try {
            String createTableSQL = "   CREATE TABLE IF NOT EXISTS user_recommendations (" +
                    "id BIGINT AUTO_INCREMENT PRIMARY KEY," +
                    " user_id BIGINT NOT NULL," +
                    "  item_id BIGINT NOT NULL," +
                    "  predicted_rating DECIMAL(10,6) NOT NULL," +
                    "  ranking INT NOT NULL," +
                    "   create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP," +
                    "   update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP," +
                    "    UNIQUE KEY uk_user_item (user_id, item_id)," +
                    "     INDEX idx_user_id (user_id)," +
                    "     INDEX idx_item_id (item_id)," +
                    "     INDEX idx_ranking (ranking)" +
                    "  ) ";

            // 使用JDBC直接执行SQL
            executeSQL(dbConfig, createTableSQL);
            log.info("推荐结果表创建/验证完成");

        } catch (Exception e) {
            log.warn("创建推荐表失败: {}", e.getMessage());
        }
    }

    /**
     * 执行SQL语句
     */
    private void executeSQL(DatabaseConfig dbConfig, String sql) {
        try (Connection conn = DriverManager.getConnection(dbConfig.getUrl(), dbConfig.getUsername(), dbConfig.getPassword()); Statement stmt = conn.createStatement()) {

            stmt.execute(sql);

        } catch (Exception e) {
            throw new RuntimeException("执行SQL失败: " + e.getMessage(), e);
        }
    }

    /**
     * 为物品生成推荐用户
     */
    public void recommendUsersForItem(int itemId, int numRecommendations) {
        if (model == null) {
            log.info("请先训练模型！");
            return;
        }

        // 为指定物品推荐用户
        Dataset<Row> itemRecs = model.recommendForAllItems(numRecommendations);

        // 过滤出指定物品的推荐
        Dataset<Row> itemSpecificRecs = itemRecs.filter(itemRecs.col("itemId").equalTo(itemId));

        log.info("可能喜欢物品 " + itemId + " 的用户：");
        itemSpecificRecs.show(false);
    }

    /**
     * 为物品生成推荐用户并保存到MySQL
     */
    public void recommendUsersForItemAndSave(int itemId, int numRecommendations) {
        if (model == null) {
            log.info("请先训练模型！");
            return;
        }

        try {
            // 为指定物品推荐用户
            Dataset<Row> itemRecs = model.recommendForAllItems(numRecommendations);

            // 过滤出指定物品的推荐
            Dataset<Row> itemSpecificRecs = itemRecs
                    .filter(col("itemId").equalTo(itemId));

            // 解析推荐结果并写入数据库
            saveItemRecommendationsToMySQL(itemSpecificRecs, itemId);

        } catch (Exception e) {
            log.error("为物品 {} 生成用户推荐失败: {}", itemId, e.getMessage(), e);
        }
    }

    /**
     * 将物品推荐结果保存到MySQL数据库
     */
    private void saveItemRecommendationsToMySQL(Dataset<Row> recommendations, int itemId) {
        try {
            // 读取数据库配置
            DatabaseConfig dbConfig = YamlConfigReader.loadDatabaseConfig("application.yml");

            // 创建物品推荐结果表（如果不存在）
            createItemRecommendationTableIfNotExists(dbConfig);

            // 转换推荐结果格式
            Dataset<Row> formattedRecs = formatItemRecommendations(recommendations, itemId);

            // 写入MySQL
            formattedRecs.write()
                    .format("jdbc")
                    .option("url", dbConfig.getUrl())
                    .option("dbtable", "item_recommendations")
                    .option("user", dbConfig.getUsername())
                    .option("password", dbConfig.getPassword())
                    .option("driver", dbConfig.getDriver())
                    .mode(SaveMode.Append)
                    .save();

            log.info("物品 {} 的 {} 条用户推荐结果已保存到MySQL", itemId, formattedRecs.count());

        } catch (Exception e) {
            log.error("保存物品推荐结果到MySQL失败: {}", e.getMessage(), e);
            throw new RuntimeException("保存物品推荐结果失败", e);
        }
    }

    /**
     * 转换物品推荐结果格式
     */
    private Dataset<Row> formatItemRecommendations(Dataset<Row> recommendations, int itemId) {
        // 创建窗口用于排名
        WindowSpec windowSpec = Window.partitionBy("itemId")
                .orderBy(col("predictedRating").desc());

        return recommendations
                .select(
                        col("itemId"),
                        explode(col("recommendations")).as("rec")
                )
                .select(
                        col("itemId"),
                        col("rec.userId").as("userId"),
                        col("rec.rating").as("predictedRating"),
                        current_timestamp().as("createTime")
                )
                .withColumn("rank", row_number().over(windowSpec))  // 按评分降序排名
                .filter(col("itemId").equalTo(itemId));  // 确保只有目标物品
    }

    /**
     * 创建物品推荐结果表
     */
    private void createItemRecommendationTableIfNotExists(DatabaseConfig dbConfig) {
        try {
            String createTableSQL = " CREATE TABLE IF NOT EXISTS item_recommendations (" +
                    " id BIGINT AUTO_INCREMENT PRIMARY KEY," +
                    " item_id BIGINT NOT NULL," +
                    "  user_id BIGINT NOT NULL," +
                    "   predicted_rating DECIMAL(10,6) NOT NULL," +
                    " ranking INT NOT NULL," +
                    "  create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP," +
                    "  update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP," +
                    "  UNIQUE KEY uk_item_user (item_id, user_id)," +
                    "  INDEX idx_item_id (item_id)," +
                    "  INDEX idx_user_id (user_id)," +
                    "    INDEX idx_ranking (ranking)," +
                    " INDEX idx_create_time (create_time)" +
                    ")  ";

            executeSQL(dbConfig, createTableSQL);
            log.info("物品推荐结果表创建/验证完成");

        } catch (Exception e) {
            log.warn("创建物品推荐表失败: {}", e.getMessage());
        }
    }

    /**
     * 为所有物品生成用户推荐并批量保存到MySQL
     */
    public void recommendUsersForAllItemsAndSave(int numRecommendations) {
        if (model == null) {
            log.info("请先训练模型！");
            return;
        }

        try {
            // 为所有物品推荐用户
            Dataset<Row> allItemRecs = model.recommendForAllItems(numRecommendations);

            log.info("开始处理 {} 个物品的用户推荐结果...", allItemRecs.count());

            // 转换格式并写入数据库
            Dataset<Row> formattedRecs = allItemRecs
                    .select(
                            col("itemId"),
                            explode(col("recommendations")).as("rec")
                    )
                    .select(
                            col("itemId").as("item_id"),
                            col("rec.userId").as("user_id"),
                            col("rec.rating").as("predicted_rating"),
                            current_timestamp().as("create_time")
                    )
                    .withColumn("ranking",
                            row_number().over(Window.partitionBy("item_id")
                                    .orderBy(col("predicted_rating").desc())));

            // 写入MySQL（覆盖模式）
            formattedRecs.write()
                    .format("jdbc")
                    .option("url", globalDbConfig.getUrl())
                    .option("dbtable", "item_recommendations")
                    .option("user", globalDbConfig.getUsername())
                    .option("password", globalDbConfig.getPassword())
                    .option("driver", globalDbConfig.getDriver())
                    .mode(SaveMode.Overwrite)
                    .save();

            log.info("成功为所有物品生成用户推荐，共 {} 条记录", formattedRecs.count());

        } catch (Exception e) {
            log.error("批量生成物品用户推荐失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 预测用户对物品的评分
     */
    public void predictRating(int userId, int itemId) {
        if (model == null) {
            log.info("请先训练模型！");
            return;
        }

        // 创建要预测的数据
        List<Row> predictionData = Collections.singletonList(RowFactory.create(userId, itemId));

        StructType schema = new StructType(new StructField[]{DataTypes.createStructField("userId", DataTypes.IntegerType, false), DataTypes.createStructField("itemId", DataTypes.IntegerType, false)});

        Dataset<Row> predictDF = spark.createDataFrame(predictionData, schema);
        Dataset<Row> predictions = model.transform(predictDF);

        log.info("用户 " + userId + " 对物品 " + itemId + " 的预测评分：");
        predictions.show();
    }

    /**
     * 保存模型
     */
    public void saveModel(String path) throws IOException {
        if (model != null) {
            model.write().overwrite().save(path);
            log.info("模型已保存到: " + path);
        }
    }


    /**
     * 从数据库读取ALS训练数据
     * 假设数据库表结构：
     * CREATE TABLE user_ratings (
     *     id BIGINT AUTO_INCREMENT,
     *     user_id INT NOT NULL,
     *     item_id INT NOT NULL,
     *     rating FLOAT NOT NULL,
     *     create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
     *     PRIMARY KEY (id)
     * );
     */

    /**
     * 加载模型
     */
    public void loadModel(String path) {
        model = ALSModel.load(path);
        log.info("模型已从 " + path + " 加载");
    }

    /**
     * 从 YAML 配置读取数据库连接信息
     */
    public Dataset<Row> readDataFromDatabase() {
        try {
            // 1. 读取 YAML 配置
            DatabaseConfig dbConfig = YamlConfigReader.loadDatabaseConfig("application.yml");

            // 2. 验证必要配置
            validateDatabaseConfig(dbConfig);

            // 3. 获取连接属性
            Properties connectionProperties = dbConfig.toProperties();

            log.info("数据库连接属性dbConfig:{}", JSONObject.toJSONString(dbConfig));
            log.info("数据库连接属性connectionProperties:{}", JSONObject.toJSONString(connectionProperties));

            // 4. 从数据库读取数据
            Dataset<Row> dataset = spark.read().jdbc(dbConfig.getUrl(), "(" + dbConfig.getQuery() + ") as ratings", connectionProperties);

            log.info("成功从数据库读取数据，数量: {}", dataset.count());

            return dataset;
        } catch (Exception e) {
            log.error("从数据库读取数据失败:{}", e.getMessage());
            e.printStackTrace();
        }

        return null;
    }

    /**
     * 验证数据库配置
     */
    private void validateDatabaseConfig(DatabaseConfig config) {
        if (config.getUrl() == null || config.getUrl().isEmpty()) {
            throw new IllegalArgumentException("数据库URL不能为空");
        }
        if (config.getUsername() == null || config.getUsername().isEmpty()) {
            throw new IllegalArgumentException("数据库用户名不能为空");
        }
        if (config.getQuery() == null || config.getQuery().isEmpty()) {
            throw new IllegalArgumentException("SQL查询语句不能为空");
        }
    }

    /**
     * 使用 Spark JDBC 选项的替代方法（更灵活）
     */
    public Dataset<Row> readDataFromDatabaseWithOptions() {
        DatabaseConfig dbConfig = YamlConfigReader.loadDatabaseConfig("application.yml");

        return spark.read().format("jdbc").
                option(JDBCOptions.JDBC_URL(), dbConfig.getUrl()).
                option(JDBCOptions.JDBC_TABLE_NAME(), "(" + dbConfig.getQuery() + ") as ratings").
                option("user", dbConfig.getUsername()).
                option("password", dbConfig.getPassword()).
                option(JDBCOptions.JDBC_DRIVER_CLASS(), dbConfig.getDriver())
                // 添加性能优化选项
                .option("fetchsize", dbConfig.getSparkJdbc() != null ? dbConfig.getSparkJdbc().getFetchSize() : 1000).option("partitionColumn", dbConfig.getSparkJdbc() != null ? dbConfig.getSparkJdbc().getPartitionColumn() : "userId").option("lowerBound", dbConfig.getSparkJdbc() != null ? dbConfig.getSparkJdbc().getLowerBound() : "1").option("upperBound", dbConfig.getSparkJdbc() != null ? dbConfig.getSparkJdbc().getUpperBound() : "1000000").option("numPartitions", dbConfig.getSparkJdbc() != null ? dbConfig.getSparkJdbc().getNumPartitions() : 4).load();
    }
}
